package com.youshang.log;

import org.apache.kafka.common.utils.ByteUtils;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;

/**
 * 解析kafka产生的日志文件，从而得到里面的数据
 * @author youshang
 * @date 2021/01/14 19:00
 **/
public class KafkaLogExample {
    public static void main(String[] args) throws IOException {
        String logDir = "/Users/youshang/Downloads/gc/00000000000000000000";
        String indexFile = logDir.concat(".index");
        String logFile = logDir.concat(".log");
//        FileChannel indexFileChannel = new FileInputStream(indexFile).getChannel();
//        FileChannel logFileChannel = new FileInputStream(logFile).getChannel();
        FileChannel indexFileChannel = new RandomAccessFile(indexFile, "rw").getChannel();
        FileChannel logFileChannel = new RandomAccessFile(logFile,"rw").getChannel();

        //获取一块内存空间
        MappedByteBuffer indexByteBuffer = indexFileChannel.map(FileChannel.MapMode.READ_ONLY, 0, indexFileChannel.size());
        MappedByteBuffer logByteBuffer = logFileChannel.map(FileChannel.MapMode.READ_ONLY, 0, indexFileChannel.size());

        while (indexByteBuffer.hasRemaining()) {
            int relativeOffset = indexByteBuffer.getInt();
            int physicalPosition = indexByteBuffer.getInt();
            if (relativeOffset == 0){
                return;
            }

            logByteBuffer.position(physicalPosition);
            long baseOffset = logByteBuffer.getLong();
            int batchLength = logByteBuffer.getInt();
            int partitionLeaderEpoch = logByteBuffer.getInt();
            int crc32 = logByteBuffer.getInt();
            byte magic = logByteBuffer.get();
            logByteBuffer.get(new byte[2]);
            int lastOffsetDelta = logByteBuffer.getInt();
            long firstTimestamp = logByteBuffer.getLong();
            long maxTimestamp = logByteBuffer.getLong();
            long producerId = logByteBuffer.getLong();
            short producerEpoch = logByteBuffer.getShort();
            int firstSequence = logByteBuffer.getInt();
            int recordsCount = logByteBuffer.getInt();

            System.out.println("baseOffset: "+baseOffset + "  batchLength: "+ batchLength + " lastOffsetDelta: "+lastOffsetDelta + " maxTimestamp: "+maxTimestamp +
            " magic: "+magic + " producerId: "+ producerId + " recordsCount: "+recordsCount);
            for (int i = 0; i < recordsCount; i++) {
                int length = ByteUtils.readVarint(logByteBuffer);
                logByteBuffer.get();
                long timestampDelta = ByteUtils.readVarlong(logByteBuffer);
                int offsetDelta = ByteUtils.readVarint(logByteBuffer);
                int keyLength = ByteUtils.readVarint(logByteBuffer);


                //-------------------
                byte[] keyBytes = new byte[keyLength];
                for (int offset = 0; offset < keyLength; offset++) {
                    byte b = logByteBuffer.get();
                    keyBytes[offset] = b;
                }

                int valueLength = ByteUtils.readVarint(logByteBuffer);
                byte[] valueBytes = new byte[valueLength];
                for (int offset = 0; offset < valueLength; offset++) {
                    byte b = logByteBuffer.get();
                    valueBytes[offset] = b;
                }

//                String string = getString(byteBuffer);
                String key = new String(keyBytes);
                String value = new String(valueBytes);
                System.out.println("key: " + key + "    value: "+value );


                int headersCount = ByteUtils.readVarint(logByteBuffer);
                for (int j = 0; j < headersCount; j++) {

                }
            }


            System.out.println("offset->" + (relativeOffset + 9897265) + ";physicalPosition->" + physicalPosition);
        }



    }


    /**
     * ByteBuffer 转换 String
     * @param buffer
     * @return
     */
    public static String getString(ByteBuffer buffer)
    {
        Charset charset = null;
        CharsetDecoder decoder = null;
        CharBuffer charBuffer = null;
        try
        {
            charset = Charset.forName("UTF-8");
            decoder = charset.newDecoder();
//             charBuffer = decoder.decode(buffer);//用这个的话，只能输出来一次结果，第二次显示为空
            charBuffer = decoder.decode(buffer.asReadOnlyBuffer());
            return charBuffer.toString();
        }
        catch (Exception ex)
        {
            ex.printStackTrace();
            return "";
        }
    }
}
